package com.bw.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.app.metrics.GoodsMetricAggregateFunction;
import com.bw.app.metrics.GoodsMetricWindowFunction;
import com.bw.app.metrics.GoodsMetricsResult;
import com.bw.common.utils.MyClickHouseUtil;
import com.bw.common.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.time.Duration;

/**
 * 商品详情页指标统计（修复收藏人数问题）
 */
public class GoodsDetailMetricsApp {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 1. 从Kafka读取action log数据
        String topic = "dwd_traffic_action_log";
        String groupId = "goods_metrics_group";
        DataStreamSource<String> kafkaStream = env.addSource(
                MyKafkaUtil.getFlinkKafkaConsumer(topic, groupId)
        );

        // 2. 数据解析和ETL
        SingleOutputStreamOperator<JSONObject> parsedStream = kafkaStream
                .map(new MapFunction<String, JSONObject>() {
                    @Override
                    public JSONObject map(String jsonStr) throws Exception {
                        try {
                            return JSON.parseObject(jsonStr);
                        } catch (Exception e) {
                            System.err.println("JSON解析错误: " + jsonStr);
                            return null;
                        }
                    }
                })
                .filter(new FilterFunction<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject obj) throws Exception {
                        return obj != null && obj.getJSONObject("common") != null;
                    }
                });

        // 3. 过滤：所有商品详情页相关的数据
        SingleOutputStreamOperator<JSONObject> relevantStream = parsedStream
                .filter(new FilterFunction<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject obj) throws Exception {
                        JSONObject page = obj.getJSONObject("page");
                        if (page == null) return false;

                        String pageId = page.getString("page_id");
                        String lastPageId = page.getString("last_page_id");

                        // 包含以下情况：
                        // 1. 当前页面是商品详情页
                        // 2. 从商品详情页跳转过来的页面（用于统计加购行为）
                        return "good_detail".equals(pageId) || "good_detail".equals(lastPageId);
                    }
                });

//        relevantStream.print("相关数据>>>");

        // 4. 分配时间戳和水位线
        SingleOutputStreamOperator<JSONObject> timedStream = relevantStream
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                                .withTimestampAssigner((event, timestamp) -> {
                                    Long ts = event.getLong("ts");
                                    return ts != null ? ts : 0L;
                                })
                );

        // 5. 应用窗口统计
        SingleOutputStreamOperator<GoodsMetricsResult> resultStream = timedStream
                .windowAll(TumblingEventTimeWindows.of(Time.minutes(1)))
                .aggregate(new GoodsMetricAggregateFunction(), new GoodsMetricWindowFunction());

        // 6. 输出结果
        resultStream.print("商品指标统计结果>>>");
        resultStream.addSink(MyClickHouseUtil.getSinkFunction("insert into dws_goods_metrics_result values(?,?,?,?,?,?)"));

        env.execute("Goods Detail Metrics Job");
    }
}